#include "../inc/rps_access.h"
#include "../inc/fdwrapper.h"
#include "../inc/rps_common.h"

static struct thread* g_threads;
static struct event_base* g_base;
static struct evconnlistener* g_listener;
static int g_last_thread = -1;
static int g_init_count = 0;

static pthread_mutex_t g_init_lock;
static pthread_cond_t g_init_cond;
static pthread_mutex_t g_worker_hang_lock;
static pthread_mutex_t test_lock;


static int count_debug_ = 0;

conn** conns;

int sigignore(int sig) {
    struct sigaction sa;
	sa.sa_handler = SIG_IGN;
	sa.sa_flags = 0;

    if (sigemptyset(&sa.sa_mask) == -1 || sigaction(sig, &sa, 0) == -1 ) {
        return -1;
    }
	
    return 0;
}

void conn_init() {
	conns = (conn**)calloc(max_fds, sizeof(conn*));
	if (conns == NULL) {
		fprintf(stderr, "Failed to allocate connection structures\n");
		exit(1);
	}
}

void conn::set_state(enum conn_states state) {
	this->state = state;
}

static int on_http_body(http_parser* parser, const char* at, size_t length) {
	conn* c = (conn*)parser->data;
	assert(c);
	//printf("c=%d, c->events=%d\n", c->sfd, c->ev_flags);
	printf("Body: %.*s\n", (int)length, at);
	
	int ret = 0;
//	pthread_mutex_lock(&test_lock_2);
	Json::Reader reader;
	Json::Value table = Json::Value::null;
	if (reader.parse(at, table) == false) {
		fprintf(stderr, "parse body met error:is not a json.\n" );
		return ret;
	}
//	pthread_mutex_unlock(&test_lock_2);
	if (table.isObject() 
		&& table.isMember("AgentProtocol") 
		&& table["AgentProtocol"].isMember("Header") 
		&& table["AgentProtocol"]["Header"].isMember("MessageType")) {
		std::string req_type = table["AgentProtocol"]["Header"]["MessageType"].asCString();
		std::string uuid = table["AgentProtocol"]["Body"]["SerialNumber"].asCString();
		if (req_type == DEVICE_REGISTER) {
			fprintf(stderr, "[INFO]: DEVICE REGISTER !\n");
			std::string area = table["AgentProtocol"]["Body"]["Area"].asCString();
			if (area.length() == 0) {
				area = "Default:Default:Default";
			}
			
			std::string oemid = table["AgentProtocol"]["Body"]["RewriteOemID"].asCString();
			if (oemid.length() == 0) {
				oemid = "General";
			}
			
			std::string dev_port = table["AgentProtocol"]["Body"]["DevicePort"].asCString();
			if (dev_port.length() == 0) {
				dev_port = "-1";
			}
			
			std::string req_type = table["AgentProtocol"]["Header"]["MessageType"].asCString();
			std::string uuid = table["AgentProtocol"]["Body"]["SerialNumber"].asCString();
			
			std::cout << req_type << std::endl;
			std::cout << uuid << std::endl;
			std::cout << area << std::endl;
			std::cout << oemid << std::endl;
			
		//	pthread_mutex_lock(&test_lock);
			ret = device_register(c, uuid, area, oemid, dev_port);
		//	pthread_mutex_unlock(&test_lock);
			return 0;	
		} 
		
	}
	
	return 0;
}
	
int conn::try_read_network() {
	printf("try_read_network\n");

	int ret = -1;
	int nparsed = 0;
	
	do {
		ret = read(sfd, r_curr, r_size);
	} while(ret < 0 && errno == EINTR);
		
	if (ret < 0) {
		close(sfd);
		delete_hash_table(this->uuid);
//		pthread_mutex_lock(&g_worker_hang_lock);
		if (this->event != NULL) {
			evtcp_safe_free(this->event, event_free);
		}
		
		
		fprintf(stderr, "read met error:%s, errno:%d.\n", strerror(errno), errno);
//		pthread_mutex_unlock(&g_worker_hang_lock);
		return ret;
	} else if (ret == 0) {
		close(sfd);
		delete_hash_table(this->uuid);
	//	pthread_mutex_lock(&g_worker_hang_lock);
		if (this->event != NULL) {
			evtcp_safe_free(this->event, event_free);
		}
		
		
		
		fprintf(stderr, "peer socket closed.\n");
//		pthread_mutex_unlock(&g_worker_hang_lock);
		return ret;
	} else {	
	
		
	//	count_debug_ ++;
		
		fprintf(stderr, "count_debug_=%d\n", this->ev_flags);
		
		thread->parser->data = (void*)this;
		nparsed = http_parser_execute(thread->parser, thread->settings, r_buf, ret);
		
		if (nparsed != ret) {
			fprintf(stderr, 
				"parser met error:%s (%s)\n",
				http_errno_description(HTTP_PARSER_ERRNO(thread->parser)),
				http_errno_name(HTTP_PARSER_ERRNO(thread->parser))
			);
			
			printf("===========\n");
			if (this->event != NULL) {
				evtcp_safe_free(this->event, event_free);
			}
		
			close(sfd);
			delete_hash_table(this->uuid);
			
			return ret;
		}
		
	/*
		char* p = strstr(r_buf, "\r\n\r\n");
		if (p == NULL) {
			fprintf(stderr, "post data is nil.\n");
			
			return ret;
		} else {
			p += 4;
			pthread_mutex_lock(&g_worker_hang_lock);
			count_debug_ ++;
			pthread_mutex_unlock(&g_worker_hang_lock);
			printf("recv data=%s, count=%d.\n", p, count_debug_);
			
			int succ = parse_http_body(p, this);
			if (succ != 0) {
				if (this->event != NULL) {
					evtcp_safe_free(this->event, event_free);
				}
				close(sfd);
				
				return ret;
			} else {
			//  We need to clean the connect r_buf and w_buf
				memset(r_buf, 0, r_size);
				memset(w_buf, 0, w_size);
				r_curr = r_buf;
				w_curr = w_buf;
				r_bytes = w_bytes = 0;	
				ret = 0;

				return ret;
			}
		}		*/
	}

	return ret;
}
	
bool conn::update_event(const int new_flags) {
	struct event_base* base = event->ev_base;
	if (base == NULL) {
		return false;
	}
	if (ev_flags == new_flags) {
		return true;
	}

	if (event != NULL) {
		evtcp_safe_free(event, event_free);
	}
	
	event = event_new(base, sfd, new_flags, event_handler, (void*)this);
	ev_flags = new_flags;
	if (event_add(event, 0) == -1) {
		perror("event_add_1");
		return false;
	}
	
	return true;
}

void conn::event_handler(const int fd, const short which, void* arg) {
	conn* c = (conn*)arg;
	assert(c != NULL);

	c->which = which;		
	if (fd != c->sfd) {
		if (c->event != NULL) {
			evtcp_safe_free(c->event, event_free);
			close(fd);
			close(c->sfd);
		}
		return;
	}

	bool stop = false;
	while(!stop) {
		switch(c->state) {			
			case conn_start:				
				if (!c->update_event( EV_READ | EV_PERSIST)) {
					c->set_state(conn_closing);
				} else {
					c->set_state(conn_read);
				}
				break;
			case conn_read:
				c->try_read_network();
				stop = true;
				break;
			case conn_closing:
				if (c->event != NULL) {
					evtcp_safe_free(c->event, event_free);
				}
				stop = true;
				break;
		}
	}
}

conn* conn::conn_new(const int sfd
					, enum conn_states init_state
					, const int event_flags
					, struct event_base* base ) {
	assert(sfd > 0 && sfd < max_fds);
	conn* c = conns[sfd];
	if (c == NULL) {
		if (!( c = (conn*)calloc(1, sizeof( conn )))) {
            fprintf(stderr, "failed to allocate connection object\n");
            return NULL;
        }
				
		c->r_buf = c->w_buf = NULL;
		
		c->r_size = 2048; //DATA_BUFFER_SIZE;
		c->w_size = 1024; //DATA_BUFFER_SIZE;
	
		c->r_buf = (char*)calloc(1, c->r_size);
		c->w_buf = (char*)calloc(1, c->w_size);
		
	
		c->sfd     = sfd;
		conns[sfd] = c;
	} 

	c->state   = init_state;
	c->r_bytes = c->w_bytes = 0;
	c->r_curr  = c->r_buf;
	c->w_curr  = c->w_buf;
	
	c->dev_port   = -1;
	c->flush_time = 0;
		
	memset(c->uuid, '\0', sizeof(c->uuid));
	memset(c->area, '\0', sizeof(c->area));
	memset(c->oemid, '\0', sizeof(c->oemid));

	c->event = event_new(base, sfd, event_flags, event_handler, (void*)c);
	if (c->event == NULL) {
		perror("event_new_2");
		exit(-1);
	}

	c->ev_flags = event_flags;
	printf("c->ev_flag=%d\n",c->ev_flags);
	if (event_add(c->event, 0) == -1) {
		printf( "c->sfd=%d\n", sfd );
		perror("event_add_2");
		return NULL;
    }

	return c;
}
	
void wait_for_thread_registration(int nthread) {
	while(g_init_count < nthread) {
		pthread_cond_wait(&g_init_cond, &g_init_lock);
	}
}

void register_thread_initialized() {
	pthread_mutex_lock(&g_init_lock);
	
	g_init_count ++;
	pthread_cond_signal(&g_init_cond);
	
	pthread_mutex_unlock(&g_init_lock);
	
	pthread_mutex_lock(&g_worker_hang_lock);
	pthread_mutex_unlock(&g_worker_hang_lock);
}

static void create_worker(void* (*func)(void*), void* args) {
	int ret = 0;
	pthread_attr_t attr;
	pthread_attr_init(&attr);
	thread* me = (thread*)args;
	
	if ((ret = pthread_create(&me->pid, &attr, func, args)) != 0) {
		fprintf(stderr, "can't create thread:%s.\n", strerror(ret));
		exit(EXIT_FAILURE); 
	}
	
	pthread_detach(me->pid);
}

void thread_process(int fd, short which, void* args) {
	printf( "run thread process\n" );
	struct thread* me = (struct thread*)args;
	
	char buf[1];
	if (read(fd, buf, 1) != 1) {
		perror("read met error");
        return;
	}

	CQ_ITEM* item = NULL;
	conn* c       = NULL;
	
	switch(buf[0]) {
		case 'c':
			item = me->new_conn_queue->pop();
			if (item == NULL) {
				break;
			}
			
			printf("item->sfd=%d\n", item->sfd);
			printf("item->ev_flags=%d\n", item->ev_flags);
			printf("item->init_state=%d\n", item->init_state);
	
			c = conn::conn_new(item->sfd, item->init_state, item->ev_flags, me->base);

			if (c == NULL) {
				fprintf(stderr, "conn_new return nil item\n");
				close(item->sfd);
			} else {
				c->thread = me;
			}
		
			me->new_conn_queue->free(item);
	}
}

void setup_thread(struct thread* me) {
	struct event_config* ev_config;
    ev_config = event_config_new();
    event_config_set_flag(ev_config, EVENT_BASE_FLAG_NOLOCK);
    me->base = event_base_new_with_config(ev_config);
    event_config_free(ev_config);
	
	if (!me->base) {
		fprintf(stderr, "can't allocate event base.\n");
		exit(EXIT_FAILURE);
	}
	
	int ev_flags = EV_READ | EV_PERSIST;
	event_set(&me->notify_event, me->notify_receive_fd, ev_flags, thread_process, me);
	event_base_set(me->base, &me->notify_event);
	
	if (event_add(&me->notify_event, 0) == -1) {
		fprintf(stderr, "can't monitor libevent notify pipe\n");
		exit(EXIT_FAILURE);
    }
	
	me->new_conn_queue = (CQ*)calloc(1, sizeof(CQ));
    if (me->new_conn_queue == NULL) {
        perror("failed to allocate memory for connection queue");
        exit(EXIT_FAILURE);
    }
	
	me->new_conn_queue->init();
	
	me->settings = (http_parser_settings*)calloc(1, sizeof(http_parser_settings));
	if (me->settings == NULL) {
		fprintf(stderr, "cat not allocate settings.\n");
		exit(EXIT_FAILURE);
	}
	
	me->settings->on_body = on_http_body;
	
	me->parser = (http_parser*)calloc(1, sizeof(http_parser));
	if (me->parser == NULL) {
		fprintf(stderr, "cat not allocate parser.\n");
		exit(EXIT_FAILURE);
	}
	
	http_parser_init(me->parser, HTTP_REQUEST);
	
	const char* redis_addr = "10.10.18.2";
	int redis_port = 6379;
	
	redis* handle = (redis*)calloc(1, sizeof(redis));
	assert(handle);
	handle->status = 0;
	handle->ctx = redisAsyncConnect(redis_addr, redis_port);
	handle->ctx->data = handle;
	handle->base = me->base;
	me->redis_conn = handle;

	redisLibeventAttach(me->redis_conn->ctx, me->redis_conn->base);
	redisAsyncSetConnectCallback(me->redis_conn->ctx, redis_conn_cb);
	redisAsyncSetDisconnectCallback(me->redis_conn->ctx, redis_disconn_cb);	

}

void* worker_process(void* args) {
	thread* me = (thread*)args;
	
	register_thread_initialized();
		
	event_base_dispatch(me->base);
	event_base_free(me->base);
	
	pthread_exit(NULL);
}

void thread_init(int nthread) {
	pthread_cond_init(&g_init_cond, NULL);
	pthread_mutex_init(&g_init_lock, NULL);
	pthread_mutex_init(&g_worker_hang_lock, NULL);
//	pthread_mutex_init(&test_lock, NULL);
//	pthread_mutex_init(&test_lock_2, NULL);
	
	g_threads = (thread*)calloc(nthread, sizeof(thread));
    if (!g_threads) {
        perror("can't allocate thread descriptors");
        exit(EXIT_FAILURE);
    }
	
	for (int i = 0; i < nthread; ++ i) {
		int fd[2] = { 0, 0 };
		if (pipe(fd) < 0) {
			fprintf(stderr, "can't create the pipe:%s.\n", strerror(errno));
			exit(-1);
		}
		
		g_threads[i].notify_receive_fd = fd[0];
		g_threads[i].notify_send_fd    = fd[1];
		
		setup_thread(&g_threads[i]);
	}
	
	for (int i = 0; i < nthread; ++ i) {
		create_worker(worker_process, &g_threads[i]);
	}
	
	pthread_mutex_lock(&g_init_lock);
	
	wait_for_thread_registration(nthread);
	
	pthread_mutex_unlock(&g_init_lock);
	
}

void dispatch_conn_new(int newfd, enum conn_states init_state, int ev_flags) {
	if (newfd <= 0) {
		fprintf(stderr, "dispatch_conn_new met error:newfd is <= 0.\n");
		return;
	}
	
	int tid       = (g_last_thread + 1) % THREAD_NUM;
	g_last_thread = tid;
	
	struct thread* thread = g_threads + tid;
	if(thread == NULL) {
		fprintf(stderr, "select thread met error:thread is nil.\n");
		exit(EXIT_FAILURE);
	}
	
	cq_item* item = (cq_item*)calloc(1, sizeof(cq_item));
	if (item == NULL) {
		fprintf(stderr, "failed to allocate memory for connection object.\n");
		close(newfd);
		return;
	}
	
	item->sfd        = newfd;
	item->init_state = init_state;
	item->ev_flags   = ev_flags;

	thread->new_conn_queue->push(item);
	
	char buf[1] = { 'c' };
	
	int ret = write(thread->notify_send_fd, buf, 1);
	if(ret != 1) {
		perror("writing to thread notify pipe");
	}
}

void evtcp_monitor_cb(struct evconnlistener* listener
					, int newfd
					, struct sockaddr* s
					, int sl, void* arg) {	
	return dispatch_conn_new(newfd, conn_start, EV_READ | EV_PERSIST);
}

int evtcp_accept_socket(evutil_socket_t sock, int backlog) {
	int err = 1;

    if (g_base == NULL || sock == -1) {
        exit(EXIT_FAILURE);
    }

    do {
		g_listener = evconnlistener_new(g_base
			, evtcp_monitor_cb
			, g_base
			, LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE
			, backlog
			, sock
		);
        
        if (g_listener == NULL) {
            exit(EXIT_FAILURE);
        }
		
		err = 0;
	} while(0);
	
	if (err == 1) {
		if (g_listener != NULL) {
			evconnlistener_free(g_listener);
		}	

        return -1;
    }

    return 0;
}

int evtcp_bind_addr(struct sockaddr* sa, size_t sin_len, int backlog) {
	int	fd    = 0;
    int on    = 1;
    int error = 1;

    if (g_base == NULL) {
		fprintf(stderr, "g_base is NULL.\n");
        return -1;
    }
	
	if (sigignore(SIGPIPE) == -1)  {
		perror("failed to ignore SIGPIPE.");
		exit(EXIT_FAILURE);
    }
	
	do {
		if ((fd = socket(sa->sa_family, SOCK_STREAM, 0)) == -1) {
			fprintf(stderr, "create socket met error:%s.\n", strerror(errno));
            return -1;
        }

        evutil_make_socket_closeonexec(fd);
        evutil_make_socket_nonblocking(fd);

		set_socket_linger(fd);
		set_socket_reusable(fd);
		set_socket_keepalive(fd);
		set_socket_nodelay(fd);
		
        if (sa->sa_family == AF_INET6) {
            if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &on, sizeof(on)) == -1) {
                break;
            }
        }

        if (bind(fd, sa, sin_len) == -1) {
            break;
        }

        error = 0;
    } while(0);
	
	if (error == 1) {
        if (fd != -1) {
            evutil_closesocket(fd);
        }

        return -1;
    }
	
	if (evtcp_accept_socket(fd, backlog) == -1) {
        evutil_closesocket(fd);

        return -1;
    }

    return 0;
}

int server_socket(const char* baddr, unsigned short port, int backlog) {
	conn_init();
	struct sockaddr*    sa;
    struct sockaddr_in6 sin6   = { 0 };
    struct sockaddr_in  sin    = { 0 };
    size_t              sin_len;

    if (!strncmp(baddr, "ipv6:", 5)) {
        baddr           += 5;
        sin_len          = sizeof(struct sockaddr_in6);
        sin6.sin6_port   = htons(port);
        sin6.sin6_family = AF_INET6;

        evutil_inet_pton(AF_INET6, baddr, &sin6.sin6_addr);
        sa = (struct sockaddr*)&sin6;
    } else {
        if (!strncmp(baddr, "ipv4:", 5)) {
            baddr += 5;
        }

        sin_len             = sizeof(struct sockaddr_in);
        sin.sin_family      = AF_INET;
        sin.sin_port        = htons(port);
        sin.sin_addr.s_addr = inet_addr(baddr);

        sa = (struct sockaddr*)&sin;
    }
	
	return evtcp_bind_addr(sa, sin_len, backlog);
}

int main() {	
	//Add events under different threads must be used
/*	int ret = -1;
	if (evthread_use_pthreads() != 0) {
		fprintf(stderr, "use evthread_use_pthreads error!\n");
		exit(-1);
	}
*/	
	struct event_config* ev_config;
	ev_config = event_config_new();
	event_config_set_flag(ev_config, EVENT_BASE_FLAG_NOLOCK);
	g_base = event_base_new_with_config(ev_config);
	event_config_free(ev_config);

	thread_init(THREAD_NUM);
	
	int ret = server_socket("10.2.5.51", 8088, 65535);
	if (ret == -1) {
		exit(EXIT_FAILURE);
	}

	event_base_dispatch(g_base);

	event_base_free(g_base);
	
	return 0;
}
